package com.bjtu.boatmq.client.consumer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;

public class RouterConnector {

	    private static final ThreadLocal<Socket> threadConnect = new ThreadLocal<Socket>(); 
	    
	    private static String HOST ;

	    private static int PORT ;
	    
	    private static Socket client;
	    
	    private static OutputStream outStr = null;
	    
	    private static InputStream inStr = null;
	    private static String ConsumerId;
	    private static Consumer consumer;
	    
	   
	    //收到的消息
	    private static String Message=null;
	    
	    private static Thread tRecv = new Thread(new RecvThread());
	    private static Thread tKeep = new Thread(new KeepThread());

	    public RouterConnector(Consumer consumer,String RouterHost, int RouterPort, int HeartBeat, String ConsumerId) {
			// TODO Auto-generated constructor stub
	    	this.HOST=RouterHost;
	    	this.PORT=RouterPort;
	    	this.ConsumerId=ConsumerId;
	    	this.consumer=consumer;
	    	try {
	    		
	    		RouterConnector.connect();
	    		tRecv.start();
	    		
	    	} catch (UnknownHostException e) {
	    			e.printStackTrace();
	    			} catch (IOException e) {
	    				e.printStackTrace();
	    				}
	    }
	    
	    public String getMsg() {
	    	return Message;
	    	
	    }
	    
		public static void connect() throws UnknownHostException, IOException  {
	        client = threadConnect.get();
	        if(client == null){
	            client = new Socket(HOST, PORT);
	            threadConnect.set(client);
	            tKeep.start();
	            System.out.println("========连接开始！========");
	        }
	        outStr = client.getOutputStream();
	        inStr = client.getInputStream();
	        
	        //向Router注册ID
	        System.out.println("注册消费者节点...");
            outStr.write(RouterConnector.ConsumerId.getBytes());
            
           //璁㈤槄topic
            System.out.println("订阅Topic...");
            //consumer.Subscribe();
            outStr.write("Subscribe:boatmq".getBytes());
            
  
	    }
	    
	    public static void disconnect() {
	        try {
	            outStr.close();
	            inStr.close();
	            client.close();
	        } catch (IOException e) {
	            e.printStackTrace();
	        }

	    }
	    
	    
	    //返回从router收到的消息
	    
	    
	    private static class KeepThread implements Runnable {
	        public void run() {
	            try {
	                System.out.println("=====================开始发送心跳包==============");
	                while (true) {
	                    try {
	                        Thread.sleep(100);
	                    } catch (InterruptedException e) {
	                        // TODO Auto-generated catch block
	                        e.printStackTrace();
	                    }
	                    System.out.println("发送心跳数据包");
	                    outStr.write("send heart beat data package !".getBytes());
	                }
	            } catch (IOException e) {
	                e.printStackTrace();
	            }

	        }
	    }
	    
	    private static class RecvThread implements Runnable {
	        public void run() {
	            try {
	                System.out.println("==============开始接收数据===============");
	                while (true) {
	                    byte[] b = new byte[1024];
	                    
	                    //未知元数据格式
	                    int r = inStr.read(b);
	                    if(r>-1){
	                        String str = new String(b);
	                        Message=str;
	                        System.out.println("==============收到数据：===============");
	                        System.out.println( str );
	                        System.out.println("==============接收完成===============");
	                   
	                    }
	                }
	                
	            } catch (IOException e) {
	                e.printStackTrace();
	            }

	        }
	    

	    }
}
	
